package com.it.bigdata

import com.it.bigdata.course.utils.{DataUtil, JedisUtil}
import kafka.serializer.StringDecoder
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis
import utils.IpUtil

object RealTimeComputing {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("ealTimeComputing").setMaster("local[2]")

    val ssc = new StreamingContext(conf, Seconds(3))

    val sc: SparkContext = ssc.sparkContext

    val rdd: RDD[String] = sc.textFile("H:\\datatest\\ips\\ip.txt")

    //对rdd进行数据清洗
    val rdd2: RDD[(String, String, String)] = rdd.map(_.split("\\|")).map(ary => (ary(2), ary(3), (ary(4) + "-" + ary(5) + "-" + ary(6) + "-" + ary(7))))

    //收集
    val array: Array[(String, String, String)] = rdd2.collect()

    //广播变量
    val broadCast: Broadcast[Array[(String, String, String)]] = sc.broadcast(array)

    val kafkaParams = Map("metadata.broker.list" -> "node3:9092,node4:9092,node5:9092")

    val topics = Set("hadoop")

    val ds1: InputDStream[(String, String)] = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)

    //取出每一条数据
    val ds2: DStream[String] = ds1.map(_._2)

    //数据清洗 2019-02-28,111.177.176.10,www.search.yahoo.com,xiaodang,vip,101
    val ds3: DStream[(String, String, String, String, String, String)] = ds2.map(_.split(",")).map(arr => (DataUtil.dataTimeToDtataString(arr(0)), arr(1), arr(2).split("/")(2), arr(2).split("/")(4), arr(2).split("/")(5), arr(2).split("/")(6).split("\\.")(0)))

    //foreachRDD允许将数据发送到外部系统.如数据库
    ds3.foreachRDD(rdd => {
      //每个分区调用一次
      rdd.foreachPartition(it => {
        for (elem <- it) {
          val ipLong: Long = IpUtil.ipToLong(elem._2)

          val address: String = IpUtil.searchIpRange(ipLong, broadCast.value)

          val jedis: Jedis = JedisUtil.getJedis()

          //存地址["20190228ips",{("address",20),("address",20),("address",20)}]
          if (!jedis.hexists(elem._1 + "-ips", address)) {

            jedis.hset(elem._1 + "-ips", address, 1.toString)
          } else {

            jedis.hset(elem._1 + "-ips", address, (jedis.hget(elem._1 + "-ips", address).toInt + 1).toString)
          }
          //存搜索引擎["20190228-search",{("www.baidu.com",20),("www.sougou.com",20)s}]
          if (!jedis.hexists(elem._1 + "-search", elem._3)) {

            jedis.hset(elem._1 + "-search", elem._3, 1.toString)
          } else {
            jedis.hset(elem._1 + "-search", elem._3, (jedis.hget(elem._1 + "-search", elem._3).toInt + 1).toString)
          }
          //存老师["20190228-teacher",{("laoli",20),("laowang",20)}]
          if (!jedis.hexists(elem._1 + "-teacher", elem._4)) {
            jedis.hset(elem._1 + "-teacher", elem._4, 1.toString)
          } else {
            jedis.hset(elem._1 + "-teacher", elem._4, (jedis.hget(elem._1 + "-teacher", elem._4).toInt + 1).toString)
          }
          //存课程["20190228-course",{("100",20),("l01",20)}]
          if(!jedis.hexists(elem._1+"-course",elem._6)){
            jedis.hset(elem._1+"-course",elem._6,1.toString)
          }else{
            jedis.hset(elem._1+"-course",elem._6,(jedis.hget(elem._1+"-course",elem._6).toInt+1).toString)
          }
          JedisUtil.closeJedis(jedis)
        }
      })
    })
    ssc.start()
    ssc.awaitTermination()
  }
}
